package com.tuniu.agents.common.logger;

import com.tuniu.agents.message.MessageSender;
import com.tuniu.agents.message.StreamMessage;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class LoggingMessageSender implements MessageSender {

    @Override
    public void send(String conId, String message) {
        log.info(conId + ":" + message);
    }

    @Override
    public void send(String conId, String respId, String message) {
        log.info(conId + ":" + respId + ":" + message);
    }

    @Override
    public void send(String conId, StreamMessage message) {
        log.info(conId + ":" + message.toEventString());
    }

    @Override
    public void send(String conId, String respId, StreamMessage message) {
        log.info(conId + ":" + respId + ":" + message.toEventString());
    }

    @Override
    public void send(String conId, Flux<String> message) {
        // Handle Flux stream without response ID
        AtomicInteger messageCount = new AtomicInteger(0);

        message
                .doOnNext(msg -> {
                    int count = messageCount.incrementAndGet();
                    log.info("Stream message[{}] - Session ID: {}, Content: {}", count, conId, msg);
                })
                .doOnComplete(() -> {
                    log.info("Stream message completed - Session ID: {}, Total messages: {}", conId, messageCount.get());
                })
                .doOnError(error -> {
                    log.error("Stream message error - Session ID: {}, Error: {}", conId, error.getMessage(), error);
                })
                .subscribe();
    }

    @Override
    public void send(String conId, String respId, Flux<String> message) {
        // Handle Flux stream with response ID
        AtomicInteger messageCount = new AtomicInteger(0);

        message
                .doOnNext(msg -> {
                    int count = messageCount.incrementAndGet();
                    log.info("Stream message[{}] - Session ID: {}, Response ID: {}, Content: {}", count, conId, respId, msg);
                })
                .doOnComplete(() -> {
                    log.info("Stream message completed - Session ID: {}, Response ID: {}, Total messages: {}", conId, respId, messageCount.get());
                })
                .doOnError(error -> {
                    log.error("Stream message error - Session ID: {}, Response ID: {}, Error: {}", conId, respId, error.getMessage(), error);
                })
                .subscribe(); // Trigger stream processing
    }

    @Override
    public String getGlobalVariable(int type) {
        return "";
    }

}
